package gu.simplemq.redis;

import java.lang.reflect.Type;
import static com.google.common.base.Preconditions.*;

import com.google.common.collect.HashBasedTable;
import com.google.common.collect.Table;

import gu.simplemq.Channel;
import gu.simplemq.Constant;
import gu.simplemq.pool.BaseMQInstance;

/**
 * redis对象工厂类用于获取producer/consumer,publisher/subscriber,table对象<br>
 * 应用程序结束时会自动清除所有subscriber和consumer
 * @author guyadong
 *
 */
public class RedisFactory implements Constant {

	private RedisFactory() {}
	@SuppressWarnings("rawtypes")
	private static final Table<JedisPoolLazy,String,RedisTable> TABLE = HashBasedTable.create();
	public static<V> RedisTable<V> getTable(Class<V> clazz){
		return getTable((Type)clazz,
				JedisPoolLazy.getDefaultInstance(),
				checkNotNull(clazz,"clazz is null").getSimpleName());
	}
	public static<V> RedisTable<V> getTable(Class<V> clazz,JedisPoolLazy pool, String tablename){
		return getTable((Type)clazz,pool,tablename);
	}
	public static<V> RedisTable<V> getTable(Channel<V> channel,JedisPoolLazy pool){
		return getTable(channel.type,pool,channel.name);
	}	
	/**
	 * 返回 {@link JedisPoolLazy}对应的{@link RedisTable}实例,如果{@link #TABLES}没有找到，
	 * 就创建一个新实例并加入{@link #TABLES}
	 * @param type
	 * @param pool
	 * @param tablename
	 * @return
	 */
	@SuppressWarnings({ "rawtypes", "unchecked" })
	public static <V> RedisTable<V> getTable(Type type,JedisPoolLazy pool, String tablename){
		checkArgument(null != type && null != pool && null != tablename,"input aruments must not be null");
		// Double Checked Locking
		if(!TABLE.contains(pool, tablename) ){
			synchronized(TABLE){
				if(!TABLE.contains(pool, tablename) ){
					@SuppressWarnings("unused")
					RedisTable previous = TABLE.put(pool, tablename, new RedisTable(type,pool,tablename));
				}
			}
		}
		RedisTable table = TABLE.get(pool,tablename);
		checkState(table.getType().equals(type),"mismatch type " + type + " vs " + table.getType());
		return table;
	}
	
	private static  final BaseMQInstance<RedisConsumer,JedisPoolLazy> CONSUMERS = new BaseMQInstance<RedisConsumer,JedisPoolLazy>(){
		@Override
		protected void beforeDelete(RedisConsumer r) {
			r.close();
	}};
	private static final BaseMQInstance<RedisSubscriber,JedisPoolLazy> SUBSCRIBERS = new BaseMQInstance<RedisSubscriber,JedisPoolLazy>(){
		@Override
		protected void beforeDelete(RedisSubscriber r) {
			r.close();
	}};
	private static final BaseMQInstance<RedisProducer,JedisPoolLazy> PRODUCERS = new BaseMQInstance<RedisProducer,JedisPoolLazy>(){};
	private static final BaseMQInstance<RedisPublisher,JedisPoolLazy> PUBLISHERS = new BaseMQInstance<RedisPublisher,JedisPoolLazy>(){};
	/**
	 * 返回 {@link JedisPoolLazy}对应的{@link RedisConsumer}实例
	 * @param jedisPoolLazy
	 * @return
	 * @see gu.simplemq.redis.RedisFactory.BaseMQInstance#getInstance(gu.simplemq.redis.JedisPoolLazy)
	 */
	public static RedisConsumer getConsumer(JedisPoolLazy jedisPoolLazy) {
		return CONSUMERS.getInstance(jedisPoolLazy);
	}
	
	/** 
	 * 返回{@link JedisPoolLazy}默认实例对应的{@link RedisConsumer}实例
	 * @see  {@link JedisPoolLazy#getDefaultInstance()}
	 */
	public static RedisConsumer getConsumer() {
		return CONSUMERS.getInstance(JedisPoolLazy.getDefaultInstance());
	}
	/**
	 * 返回 {@link JedisPoolLazy}对应的{@link RedisSubscriber}实例
	 * @param jedisPoolLazy
	 * @return
	 * @see gu.simplemq.redis.RedisFactory.BaseMQInstance#getInstance(gu.simplemq.redis.JedisPoolLazy)
	 */
	public static RedisSubscriber getSubscriber(JedisPoolLazy jedisPoolLazy) {
		return SUBSCRIBERS.getInstance(jedisPoolLazy);
	}
	/** 
	 * 返回{@link JedisPoolLazy}默认实例对应的{@link RedisSubscriber}实例
	 * @see  {@link JedisPoolLazy#getDefaultInstance()}
	 */
	public static RedisSubscriber getSubscriber() {
		return SUBSCRIBERS.getInstance(JedisPoolLazy.getDefaultInstance());
	}
	/**
	 * 返回 {@link JedisPoolLazy}对应的{@link RedisProducer}实例
	 * @param jedisPoolLazy
	 * @return
	 * @see gu.simplemq.redis.RedisFactory.BaseMQInstance#getInstance(gu.simplemq.redis.JedisPoolLazy)
	 */
	public static RedisProducer getProducer(JedisPoolLazy jedisPoolLazy) {
		return PRODUCERS.getInstance(jedisPoolLazy);
	}
	/** 
	 * 返回{@link JedisPoolLazy}默认实例对应的{@link RedisProducer}实例
	 * @see  {@link JedisPoolLazy#getDefaultInstance()}
	 */
	public static RedisProducer getProducer() {
		return PRODUCERS.getInstance(JedisPoolLazy.getDefaultInstance());
	}
	/**
	 * 返回 {@link JedisPoolLazy}对应的{@link RedisPublisher}实例
	 * @param jedisPoolLazy
	 * @return
	 * @see gu.simplemq.redis.RedisFactory.BaseMQInstance#getInstance(gu.simplemq.redis.JedisPoolLazy)
	 */
	public static RedisPublisher getPublisher(JedisPoolLazy jedisPoolLazy) {
		return PUBLISHERS.getInstance(jedisPoolLazy);
	}
	/** 
	 * 返回{@link JedisPoolLazy}默认实例对应的{@link RedisPublisher}实例
	 * @see  {@link JedisPoolLazy#getDefaultInstance()}
	 */
	public static RedisPublisher getPublisher() {
		return PUBLISHERS.getInstance(JedisPoolLazy.getDefaultInstance());
	}
	/**
	 * 关闭并删除所有consumer,subscriber
	 */
	public synchronized static void closeAll(){
		CONSUMERS.clearInstances();
		SUBSCRIBERS.clearInstances();
	}
	static {
		// JVM 结束时自动清除所有consumer和subscriber对象
		Runtime.getRuntime().addShutdownHook(new Thread(){

			@Override
			public void run() {
				try {
					logger.info("RedisFactory close all of consumers,subscribers");
					closeAll();
				} catch (Throwable e) {
					logger.error(e.getMessage(),e);
				}
			}			
		});
	}
}
